package top.lyunk.demo.flink.job1;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * <pre>
 * 说明：
 * </pre>
 *
 * @author lyk
 * @date 2022/9/6
 */
public class job1 {
    public static void main(String[] args) throws Exception {
        /*
        1. 获取一个执行环境
         */
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /*
        2. 加载/创建初始数据
         */
        DataStreamSource<String> input = env.socketTextStream("127.0.0.1", 9999);
        // DataStreamSource<String> input = env.fromElements("aaa,bbb,ccc", "ddd,aaa");

        /*
        3. 指定数据的相关转换
         */
        SingleOutputStreamOperator<Tuple2<String, Integer>> result = input.flatMap(new WordCountFlatMapFunc());
        // 按key分组
        SingleOutputStreamOperator<Tuple2<String, Integer>> dataStream = result.keyBy(value -> value.f0)
                // .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .sum(1);

        dataStream.print();
        env.execute("Word Count");
    }

    public static class WordCountFlatMapFunc implements FlatMapFunction<String, Tuple2<String, Integer>>{
        @Override
        public void flatMap(String in, Collector<Tuple2<String, Integer>> out) throws Exception {
            for (String word: in.split(",")) {
                out.collect(new Tuple2<>(word, 1));
            }
        }
    }
}
